package cn.ivicar.cloud.mqtt;

import java.net.URISyntaxException;

import cn.ivicar.cloud.utils.IovProperties;
import cn.ivicar.cloud.engine.RuleEngine;

import org.apache.log4j.Logger;
import org.fusesource.mqtt.client.Future;
import org.fusesource.mqtt.client.FutureConnection;
import org.fusesource.mqtt.client.MQTT;
import org.fusesource.mqtt.client.Message;
import org.fusesource.mqtt.client.QoS;
import org.fusesource.mqtt.client.Topic;

public class MqttClient {
    private static MQTT mqtt = null;
    private static FutureConnection connection = null;

    private static Logger LOGGER = Logger.getLogger(MqttClient.class);

    private final static int SEND_BUFFER_SIZE = 2 * 1024 * 1024; // 发送最大缓冲为2M
    private final static long RECONNECTION_ATTEMPT_MAX = 6;
    private final static long RECONNECTION_DELAY = 2000;
    private static String CONNECTION_STRING = "tcp://127.0.0.1:61613";
    private static String USERNAME = "admin";
    private static String PASSWORD = "password";
    private final static boolean CLEAN_START = true;
    private static short KEEP_ALIVE = 30; // 低耗网络，但是又需要及时获取数据，心跳30s

    static {
        CONNECTION_STRING = IovProperties.getProperties().getProperty("mqtt.host");
        USERNAME = IovProperties.getProperties().getProperty("mqtt.username");
        PASSWORD = IovProperties.getProperties().getProperty("mqtt.password");
        KEEP_ALIVE = Short.parseShort(IovProperties.getProperties().getProperty("mqtt.heart"));
    }

    public static void connect() {
        mqtt = new MQTT();
        try {
            // 设置mqtt broker的IP和端口
            mqtt.setHost(CONNECTION_STRING);
            mqtt.setUserName(USERNAME);
            mqtt.setPassword(PASSWORD);

            // 连接前清空会话信息
            mqtt.setCleanSession(CLEAN_START);
            // 设置重新连接的次数
            mqtt.setReconnectAttemptsMax(RECONNECTION_ATTEMPT_MAX);
            // 设置重连的间隔时间
            mqtt.setReconnectDelay(RECONNECTION_DELAY);
            // 设置心跳时间
            mqtt.setKeepAlive(KEEP_ALIVE);
            // 设置缓冲的大小
            mqtt.setSendBufferSize(SEND_BUFFER_SIZE);

            // 获取mqtt的连接对象BlockingConnection
            connection = mqtt.futureConnection();
            connection.connect();
        } catch (URISyntaxException e) {
            LOGGER.error("MQTT连接失败", e);
        } catch (Exception e) {
            LOGGER.error("MQTT连接失败", e);
        }
    }

    public static void processMessage(Topic[] topic) {
        connection.subscribe(topic);
        while (true) {
            String msg = null;
            try {
                Future<Message> futureMessage = connection.receive();
                Message message = futureMessage.await();
                message.ack();
                String msgTopic = message.getTopic();
                msg = new String(message.getPayload());
                LOGGER.info(String.format("Message Received, topic [ %s ], message [ %s ]", msgTopic, msg));

                RuleEngine.process(msgTopic, msg);
            } catch (Exception e) {
                LOGGER.error(String.format("消息处理出现未知异常, message:%s, error:%s", msg, e.getMessage(), e));
            }
        }
    }

    public static void sendToMqtt(String topic, String msg) throws Exception {
        connection.publish(topic, msg.getBytes(), QoS.EXACTLY_ONCE, false);
        LOGGER.info(String.format("Message Published, topic [ %s ], message [ %s ]", topic, msg));
    }

    public static void tearDown() {
        if (connection != null && connection.isConnected()) {
            connection.disconnect();
        }
    }
}
